package com.bigdata.warehouse.ads;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: com.bigdata.warehouse.ads
 * @Author: zhangleqing
 * @CreateTime: 2025-08-23  15:35
 * @Description: TODO
 * @Version: 1.0
 */
public class Car {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 每天车辆进出表 dws_car_behavior  kafka
        tEnv.executeSql(
                "CREATE TABLE dws_car_behavior (\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    type STRING,\n" +
                        "    owerId INT,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    community_id INT,\n" +
                        "    communityName STRING,\n" +
                        "    city_id INT,\n" +
                        "    district_id INT,\n" +
                        "    district_name STRING,\n" +
                        "    car_in_cnt INT,\n" +
                        "    car_out_cnt INT,\n" +
                        "    stt TIMESTAMP(3),\n" +
                        "    edt TIMESTAMP(3),\n" +
                        "    ts TIMESTAMP_LTZ(3),\n" +
                        "    PRIMARY KEY (carCode) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "'connector' = 'upsert-kafka',\n" +
                        "'topic' = 'dws_car_behavior',\n" +
                        "'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "'key.format' = 'json',\n" +
                        "'value.format' = 'json'\n" +
                        ")"
        );

/*
创建Doris表
CREATE TABLE dws_car_behavior_doris (
    carCode STRING,
    carColor STRING,
    type STRING,
    owerId INT,
    owerSex STRING,
    owerAge INT,
    owerHomeplace STRING,
    owerWorkplace STRING,
    community_id INT,
    communityName STRING,
    city_id INT,
    district_id INT,
    district_name STRING,
    car_in_cnt INT,
    car_out_cnt INT,
    stt TIMESTAMP(3),
    edt TIMESTAMP(3),
    ts TIMESTAMP_LTZ(3),
    PRIMARY KEY (carCode) NOT ENFORCED
) WITH (
    'connector' = 'doris',
    'fenodes' = 'node102:8030',
    'table.identifier' = 'sca_doris_export.dws_car_behavior',
    'username' = 'root',
    'password' = '123456',
    'sink.buffer-flush.max-rows' = '1',
    'sink.buffer-flush.interval' = '1000',
    'sink.max-retries' = '3',
    'sink.enable-delete' = 'true',
    'sink.properties.format' = 'json',
    'sink.enable.batch-mode' = 'true'
)
 */
        tEnv.executeSql(
                "CREATE TABLE dws_car_behavior_doris (\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    type STRING,\n" +
                        "    owerId INT,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    community_id INT,\n" +
                        "    communityName STRING,\n" +
                        "    city_id INT,\n" +
                        "    district_id INT,\n" +
                        "    district_name STRING,\n" +
                        "    car_in_cnt INT,\n" +
                        "    car_out_cnt INT,\n" +
                        "    stt TIMESTAMP(3),\n" +
                        "    edt TIMESTAMP(3),\n" +
                        "    ts TIMESTAMP_LTZ(3),\n" +
                        "    PRIMARY KEY (carCode) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris_export.dws_car_behavior',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        // 1. 替代 sink.batch.size：达到1条数据就触发写入（测试用，生产可改1000）
                        "    'sink.buffer-flush.max-rows' = '1',\n" +
                        // 2. 替代 sink.batch.interval：1秒超时触发写入（防止数据积压）
                        "    'sink.buffer-flush.interval' = '1000',\n" +
                        // 3. 替代 sink.max-retries：失败重试3次（原参数支持，保留）
                        "    'sink.max-retries' = '3',\n" +
                        // 4. 适配 Upsert 语义：允许删除/更新（旧版通过该参数支持）
                        "    'sink.enable-delete' = 'true',\n" +
                        // 5. 数据格式：与Kafka输出匹配（原参数支持，保留）
                        "    'sink.properties.format' = 'json',\n" +
                        // 6. （可选）批量写入模式（提升性能，旧版支持）
                        "    'sink.enable.batch-mode' = 'true'\n" +
                        ")"
        );


        // 读取Kafka中的数据，插入到Dori表中
        tEnv.executeSql(
                "insert into dws_car_behavior_doris " +
                        "select * from dws_car_behavior"
        );


        // 查询Doris中是否写入成功
        //tEnv.executeSql("select * from dws_car_behavior limit 10").print();


        //env.execute("Car");
    }
}
